/**
 * @(#)Server.java, Nov 12, 2012. 
 * 
 * Copyright 2012 Yodao, Inc. All rights reserved.
 * YODAO PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 */
package com.buaaer.producerconsumer;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 *
 * @author wangheng
 *
 */
public class Server {

    private int port;
    
    private ServerSocket serverSocket;
    
    private LinkedBlockingQueue<Socket> requestQueue = new LinkedBlockingQueue<Socket>();
    
    public Server(int port) {
        this.port = port;
    }
    
    public void start() throws IOException, ClassNotFoundException {
        
        this.serverSocket = new ServerSocket(port);
        
        Consumer consumer = new Consumer(requestQueue);
        consumer.start();
        
        while(true) {
            System.out.println("server wait...");
            Socket socket = serverSocket.accept();
            System.out.println("add to queue");
            requestQueue.add(socket);
            
        }
        
    }
    
    public static void main(String[] args) throws Exception {
        Server server = new Server(10000);
        server.start();
    }
    
    class Consumer extends Thread {

        private LinkedBlockingQueue<Socket> requestQueue;
        
        private ExecutorService executors;
        
        private final static int DEFAULT_PROCESS_THREAD_NUM = 4;
        
        public Consumer(LinkedBlockingQueue<Socket> requestQueue) {
            this(requestQueue, DEFAULT_PROCESS_THREAD_NUM);
        }
        
        public Consumer(LinkedBlockingQueue<Socket> requestQueue, int processThreadNum) {
            this.requestQueue = requestQueue;
            this.executors = Executors.newFixedThreadPool(processThreadNum);
            
        }
        
        @Override
        public void run() {
            try {
                while(true) {
                    System.out.println("consumer wait...");
                    Socket socket = requestQueue.take();
                    this.executors.execute(new ProcessUnit(socket));
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            
        }
        
        class ProcessUnit implements Runnable {
            
            Socket socket;
            
            ProcessUnit(Socket socket) {
                this.socket = socket;
            }

            @Override
            public void run() {
                try {
                    System.out.println("process socket:" + socket);
                    ObjectInputStream in = new ObjectInputStream(socket.getInputStream());
                    Object object = in.readObject();
                    
                    System.out.println("client message:" + object);
                    
                    
                    ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
                    out.writeObject("echo: from server " + object);
                    System.out.println("echo: from server " + object);
                    
                    in.close();
                    out.close();
                    socket.close();
                    
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (ClassNotFoundException e) {
                    e.printStackTrace();
                }
                
            }
            
        }
        
        
    }
}

